#pragma once
#include "../common/message.hpp"
#include "../common/net.hpp"
#include "../common/dispatcher.hpp"
#include "requestor.hpp"
#include "rpc_caller.hpp"
#include "rpc_registry.hpp"
#include "rpc_topic.hpp"

namespace myrpc {
    namespace client {
        // 服务注册客户端
        class RegistryClient {
        public:
            using ptr = std::shared_ptr<RegistryClient>;
            RegistryClient(const std::string &ip, int port)
            :_requestor(std::make_shared<Requestor>()),
            _provider(std::make_shared<client::Provider>(_requestor)),
            _dispatcher(std::make_shared<Dispatcher>())
            {
                // 服务注册响应注册到Dispatcher事件分发器中
                auto rsp_cb=std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MsgType::RSP_SERVICE, rsp_cb);
                // 再将Dispatcher事件分发器对应的消息回调处理设置到底层客户端的系统回调中
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _client = ClientFactory::create<MuduoClient>(ip, port);
                _client->setMessageCallBack(msg_cb);
                _client->connect();
            }
            bool registerMethod(const std::string &method, const Address &host){
                return _provider->registerMethod(_client->connection(), method, host);
            }

        private:
            Requestor::ptr _requestor;
            Provider::ptr _provider;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };
        
        // 服务发现客户端
        class DiscoveryClient {
        public:
            using ptr = std::shared_ptr<DiscoveryClient>;
            DiscoveryClient(const std::string &ip, const int &port, const Discoverer::CloseClientCB &cb)
            :_requestor(std::make_shared<Requestor>()),
            _discoverer(std::make_shared<Discoverer>(_requestor, cb)),
            _dispatcher(std::make_shared<Dispatcher>())
            {
                // 服务发现响应注册到Dispatcher事件分发器中
                auto rsp_cb=std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MsgType::RSP_SERVICE, rsp_cb);
                // 服务上线与下线请求的回调处理也注册到Dispatcher事件分发器中
                auto req_cb = std::bind(&Discoverer::onServiceResquest, _discoverer.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<ServiceRequest>(MsgType::REQ_SERVICE, req_cb);
                // 再将Dispatcher事件分发器对应的消息回调处理设置到底层客户端的系统回调中
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _client = ClientFactory::create<MuduoClient>(ip, port);
                _client->setMessageCallBack(msg_cb);
                _client->connect();
            }
            bool serviceDiscover(const std::string &method, Address &host){
                return _discoverer->serviceDiscover(_client->connection(), method, host);
            }

        private:
            Requestor::ptr _requestor;
            Discoverer::ptr _discoverer;
            Dispatcher::ptr _dispatcher;
            BaseClient::ptr _client;
        };

        // Rpc服务调用客户端
        class RpcClient {
        public:
            using ptr = std::shared_ptr<RpcClient>;
            // 是否开启服务发现功能 决定了传入的主机地址信息是服务中心 还是目标服务主机
            RpcClient(bool enablediscovery, const std::string &ip, const int &port)
                : _enablediscovery(enablediscovery),
                  _requestor(std::make_shared<Requestor>()),
                  _dispatcher(std::make_shared<Dispatcher>()),
                  _caller(std::make_shared<RpcCaller>(_requestor))
            {
                auto rsp_cb=std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MsgType::RSP_RPC, rsp_cb);
                // 开启服务发现功能 传入过来的地址就是服务中心的地址 需要先服务发现 然后获取Rpc客户端连接 可以提供多种服务
                if(_enablediscovery){
                    auto close_client_cb = std::bind(&RpcClient::delRpcClient, this, std::placeholders::_1);
                    _discovery_client = std::make_shared<DiscoveryClient>(ip, port, close_client_cb);
                }
                // 未开启则传入过来的直接就是目标服务主机 连接关闭前只能提供这一种服务
                else{
                    auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                    _rpc_client = ClientFactory::create<MuduoClient>(ip, port);
                    _rpc_client->setMessageCallBack(msg_cb);
                    _rpc_client->connect();
                }
            }

            bool call(const std::string &method, const Json::Value &params, Json::Value &result){
                BaseClient::ptr rpc_client = getRpcClient(method);
                if (rpc_client.get() == nullptr){
                    return false;
                }
                auto conn = rpc_client->connection();
                if(!conn){
                    client_logger->ERROR("连接无效,调用失败");
                    return false;
                }
                return _caller->call(conn, method, params, result);
            }
            bool call(const std::string &method, const Json::Value &params, RpcCaller::JsonRspFuture &result){
                BaseClient::ptr rpc_client = getRpcClient(method);
                if (rpc_client.get() == nullptr){
                    return false;
                }
                auto conn = rpc_client->connection();
                if(!conn){
                    client_logger->ERROR("连接无效,调用失败");
                    return false;
                }
                return _caller->call(conn, method, params, result);
            }
            bool call(const std::string &method, const Json::Value &params, const RpcCaller::JsonRspCallBack &user_cb){
                BaseClient::ptr rpc_client = getRpcClient(method);
                if (rpc_client.get() == nullptr){
                    return false;
                }
                auto conn = rpc_client->connection();
                if(!conn){
                    client_logger->ERROR("连接无效,调用失败");
                    return false;
                }
                return _caller->call(conn, method, params, user_cb);
            }

        private:
            // 根据方法名查找Rpc客户端连接
            BaseClient::ptr getRpcClient(const std::string &method){
                BaseClient::ptr client;
                if(_enablediscovery){
                    client_logger->DEBUG("服务发现模式已启用, 开始查找服务 '%s'", method.c_str());
                    // 如果是设置了服务发现功能 需要先进行服务发现获取可提供服务的主机地址
                    Address host;
                    bool ret = _discovery_client->serviceDiscover(method, host);
                    if (ret == false){
                        client_logger->ERROR("服务发现失败, 无法找到 '%s' 服务的提供者", method.c_str());
                        return BaseClient::ptr();
                    }
                    // 根据获取到的主机地址到连接池中查找已存在的连接 不存在则创建新的连接
                    client = getRpcClient(host);
                    if (client.get() == nullptr){
                        client_logger->INFO("连接池中无可用连接, 为 '%s' @ %s:%d 创建新连接", method.c_str(), host.first.c_str(), host.second);
                        client = newRpcClient(host);
                    }
                    else {
                        client_logger->DEBUG("从连接池中复用连接 -> %s:%d", host.first.c_str(), host.second);
                    }
                }
                else{
                    client_logger->DEBUG("独立模式, 使用预配置连接");
                    client = _rpc_client;
                }
                return client;
            }
            // 创建新的Rpc客户端连接
            BaseClient::ptr newRpcClient(const Address &host){
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                BaseClient::ptr rpc_client = ClientFactory::create<MuduoClient>(host.first, host.second);
                rpc_client->setMessageCallBack(msg_cb);
                rpc_client->connect();

                // 验证连接是否真正建立
                if (!rpc_client->connected()) {
                    client_logger->ERROR("连接建立失败: %s:%d", host.first.c_str(), host.second);
                    return BaseClient::ptr();
                }

                addRpcClient(host, rpc_client);
                client_logger->INFO("已创建并添加新连接到连接池 -> %s:%d", host.first.c_str(), host.second);
                return rpc_client;
            }
            // 根据可提供服务的主机地址到连接池中获取Rpc客户端连接
            BaseClient::ptr getRpcClient(const Address &host){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it=_rpc_client_pool.find(host);
                if (it == _rpc_client_pool.end()){
                    return BaseClient::ptr();
                }
                return it->second;
            }
            // 从连接池中新增Rpc客户端连接
            void addRpcClient(const Address &host, BaseClient::ptr &rpc_client){
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_client_pool.insert(std::make_pair(host, rpc_client));
            }
            // 从连接池中删除Rpc客户端连接
            void delRpcClient(const Address &host){
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _rpc_client_pool.find(host);
                if (it != _rpc_client_pool.end()) {
                    _rpc_client_pool.erase(it);
                    client_logger->INFO("已从连接池中移除连接 -> %s:%d", host.first.c_str(), host.second);
                }
            }

        private:
            struct AddressHash{
                size_t operator()(const Address &host) const {
                    std::string add = host.first + std::to_string(host.second);
                    return std::hash<std::string>{}(add);
                }
            };
 
            bool _enablediscovery;                        // 用来标记当前Rpc客户端是否具有服务发现功能
            DiscoveryClient::ptr _discovery_client;       // 服务发现客户端——只有开启服务发现功能才需要
            Requestor::ptr _requestor;
            Dispatcher::ptr _dispatcher;
            RpcCaller::ptr _caller;
            BaseClient::ptr _rpc_client;                  
            std::mutex _mutex;
            // 连接池——开启服务发现功能才需要 用来存储已建立的Rpc客户端连接
            std::unordered_map<Address, BaseClient::ptr, AddressHash> _rpc_client_pool; 
        };
        // 发布订阅客户端
        class TopicClient {
        public:
            using ptr=std::shared_ptr<TopicClient>;
            TopicClient(const std::string &ip, int port)
            :_requestor(std::make_shared<Requestor>()),
            _dispatcher(std::make_shared<Dispatcher>()),
            _topic_manager(std::make_shared<TopicManager>(_requestor))
            {
                // 注册收到主题响应后的回调处理
                auto top_rsp_cb=std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<BaseMessage>(MsgType::RSP_TOPIC, top_rsp_cb);
                // 注册收到订阅的主题消息后的回调处理
                auto pub_msg_cb = std::bind(&TopicManager::onPublish, _topic_manager.get(), std::placeholders::_1, std::placeholders::_2);
                _dispatcher->registerHandler<TopicRequest>(MsgType::REQ_TOPIC, pub_msg_cb);
                // 创建客户端 并设置消息回调处理
                _topic_client = ClientFactory::create<MuduoClient>(ip, port);
                auto msg_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);
                _topic_client->setMessageCallBack(msg_cb);
                _topic_client->connect();
            }
            // 向外——用户提供的主题操作接口
            bool topicCreate(const std::string &topic_name){
                return _topic_manager->topicCreate(_topic_client->connection(), topic_name);
            }
            bool topicRemove(const std::string &topic_name){
                return _topic_manager->topicRemove(_topic_client->connection(), topic_name);
            }
            bool topicSubscribe(const std::string &topic_name, const TopicManager::SubCallBack &cb){
                BaseConnection::ptr conn = _topic_client->connection();
                return _topic_manager->topicSubscribe(conn, topic_name, cb);
            }
            bool topicCancel(const std::string &topic_name){
                return _topic_manager->topicCancel(_topic_client->connection(), topic_name);
            }
            bool topicPublish(const std::string &topic_name, const std::string &msg){
                return _topic_manager->topicPublish(_topic_client->connection(), topic_name, msg);
            }

            void shutdown(){
                _topic_client->shutdown();
            }

        private:
            Requestor::ptr _requestor;
            Dispatcher::ptr _dispatcher;
            TopicManager::ptr _topic_manager;
            BaseClient::ptr _topic_client;                  
        };
    }
}